package com.analysis

// import语句可以出现在任何地方，而不是只能在文件顶部。import的效果从开始延伸到语句块的结束。这可以大幅减少名称冲突的可能性。
// 如果想要引入包中的几个成员，可以使用selector（选取器）：

import java.sql.Timestamp

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.api.common.typeutils.TypeSerializer
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

import scala.collection.mutable.ListBuffer
// org.apache.flink.streaming.api.scala._； 引入org.apache.flink.streaming.api.scala包内所有成员
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.{TimeWindow, Window}
import org.apache.flink.api.java.tuple.{Tuple , Tuple1}
import org.apache.flink.util.Collector;
import java.io._


/**
  * FIXME step 1  先定义输入数据的样例类
  */
case class UserBehavior(userId: Long, itemId: Long, cataoryId: Int, behavior: String, timestamp: Long)

/**
  * FIXME step2 再定义输出数据:窗口聚合结果的WindowEnd的样例类
  * */
case class ItemViewCount(itemId: Long, windowEnd: Long, count: Long)


object HotItems {

  def main(args: Array[String]): Unit = {
    // 在 Scala 中，使用关键词 "var" 声明变量，使用关键词 "val" 声明常量。
    //val writer = new PrintWriter(new File(("E:\\work-code\\永辉数据中台\\永辉实时\\4-demo\\UserBehaviorAnalysis\\HotItemsAnalysis\\src\\main\\resources\\HotItems.processStream.out")))

    // - step 1 : 获得一个执行环境（ExecutionEnvironment）
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 并发数为1
    env.setParallelism(1)
    // 设置时间语义为 EventTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // - step 2 : 加载/创建初始数据 （Source）
    var inputTextFile = "E:\\work-code\\永辉数据中台\\永辉实时\\4-demo\\UserBehaviorAnalysis\\HotItemsAnalysis\\src\\main\\resources\\UserBehavior.csv"
    val dataStream = env.readTextFile(inputTextFile)
      .map( data => {
        // 先切分
        val dataArray = data.split(",")
        // 然后转换成我们想要的UserBehavior类型
          // userId: Long, itemId: Long, cataoryId: Int, behavior: String, timstamp: Long
        UserBehavior( dataArray(0).trim.toLong
          , dataArray(1).trim.toLong
          , dataArray(2).trim.toInt
          , dataArray(3).trim.toString
          , dataArray(4).trim.toLong )
      })
      // 从数据中提取某个字段作为时间戳
      // assignAscendingTimestamps: 分配升序时间戳
      .assignAscendingTimestamps(_.timestamp * 1000L)

      // 还可指定watermark，怎么指定watermark跟数据有关系

    // - step 3 : 指定转换算子操作数据（Transformation）
      // 先要做过滤（filter）

    // 下面两种写法效果一样：scala能自动推断  写法2就是写法1的省略
    //写法1： val processStream: DataStream[UserBehavior] = dataStream.filter( _.behavior == "pv" )
    //写法2：
    val processStream = dataStream
      .filter( _.behavior == "pv" )
      // 处理流程见： E:\work-code\永辉数据中台\永辉实时\4-demo\UserBehaviorAnalysis\HotItemsAnalysis\src\main\resources\DataStream窗口聚合原理.png
        .keyBy(
          /* 写法1： "itemId" 【不推荐】 ，因为要获取这个itemId太难了 ： val itemId = key.asInstanceOf[ Tuple1[ Long ] ].f0 */
           /* 写法2： _.itemId 【推荐】 */ _.itemId
        )
        // 根据 itemId 开一个滑动窗口timewindow， 长度为1小时，滑动步长为5分钟
        .timeWindow( Time.hours(1), Time.minutes(5) )
        // 聚合：第一个参数用于定义窗口聚合规则 ， 第二个参数用于定义输出数据结构
      /***
        * 【aggregate函数说明】
        *  preAggregator: AggregateFunction[T, ACC, V],
        *  windowFunction: WindowFunction[V, R, K, W]
        */
      .aggregate( /*AggregateFunction[T, ACC, V]*/ new CountAgg(), /*WindowFunction[V, R, K, W]*/ new WindowResult())  // 窗口聚合
      // 到这里就得到了【每个窗口做聚合完成之后的结果】

      // 接下来，是排序，但是排序之前要按照窗口进行分组
        .keyBy(_.windowEnd)   // 按照窗口分组

      // 排序
        .process( new TopNHotItem( 10 ))


    // - step 4 : 指定存放结果位置（Sink）
      // sink这里直接控制台输出
    //  【可以打开】 dataStream.print()

    processStream.print()

    import org.apache.flink.core.fs.FileSystem
    //dataStream.writeAsText("E:\\output.log", FileSystem.WriteMode.OVERWRITE)
    dataStream.writeAsText("E:\\work-code\\永辉数据中台\\永辉实时\\4-demo\\UserBehaviorAnalysis\\HotItemsAnalysis\\src\\main\\resources\\HotItems.dataStream.out", FileSystem.WriteMode.OVERWRITE)

    processStream.writeAsText("E:\\work-code\\永辉数据中台\\永辉实时\\4-demo\\UserBehaviorAnalysis\\HotItemsAnalysis\\src\\main\\resources\\HotItems.processStream.out", FileSystem.WriteMode.OVERWRITE)

    // - step 5:  最后一步：执行
    env.execute("hot items job")
    //writer.close()

  }




  // 自定义窗口函数，输出ItemViewCount
  /**
    * trait WindowFunction[IN, OUT, KEY, W <: Window] extends Function with Serializable {
    * 在键控（分组）窗口上求值的函数的基本接口。
    *   IN ：输入数据类型
    *   OUT： 输出数据类型
    *   KEY：
    *  注意： W <: Window 表示 Window是W的上界
    *
    */
  class  WindowResult() extends WindowFunction[/*IN ：输入数据类型*/ Long, /*OUT 输出数据类型： */ ItemViewCount, /*KEY : 【写法1：不推荐】 "itemId" 为Long类型，实际上这里要用元祖类型，这是因为 keyBy的返回结果是 JavaTuple，所以这里也要用 Tuple */ /*KEY : 【写法2：推荐】 "itemId" 为Long类型，因为是_.itemId，实际上这里要用Long类型，这是因为 keyBy的返回结果是 Long，所以这里也要用 Long */ Long, /*W <: Window]*/ TimeWindow]{
    override def apply(key: Long, window: TimeWindow, input: Iterable[Long], out: Collector[ItemViewCount]): Unit = {
      // ItemViewCount(itemId: Long, windowEnd: Long, count: Long)
      // 从key中取出itemId
      /*【写法1：不推荐】 val itemId = key.asInstanceOf[ Tuple1[ Long ] ].f0*/
      // val itemId = key.asInstanceOf[ Tuple1[ Long ] ].f0  这行代码为什么这么恶心？有没有更优雅的写法？
     /* 【写法2：推荐】val itemId = key;*/
      val itemId = key;
      //println("【WindowResult.apply】itemId: "+itemId +" , windowEnd: "+ window.getEnd + " , input: "+ input)
      out.collect( ItemViewCount(key, window.getEnd, input.iterator.next()))
    }

  }



  // 自定义预聚合函数，实现聚合函数接口
  // OUT 就是WindowResult的输入
  class  CountAgg() extends AggregateFunction[ /*IN ：输入数据类型*/ UserBehavior , /*ACC ：累加器的类型（中间聚合状态）。*/ Long , /*OUT 输出数据类型： */  Long]{

    /**
      * Adds the given input value to the given accumulator, returning the new accumulator value.
      * 将给定的输入值添加到给定的累加器，返回新的累加器值。
      *
      * <p>For efficiency, the input accumulator may be modified and returned.
      * 为了提高效率，可以修改并返回输入累加器。
      *
      * @param value       The value to add   要添加的值
      * @param accumulator The accumulator to add the value to    要向其中添加值的累加器
      */
    override def add(value: UserBehavior, accumulator: Long): Long = {
      //println("【add】accumulator + 1: "+accumulator + 1)
      accumulator + 1
    }

    /**
      * Creates a new accumulator, starting a new aggregate.
      * 创建新的累加器，开始新的聚合。
      *
      * <p>The new accumulator is typically meaningless unless a value is added via {@link #add(Object, Object)}.
      * 除非通过{@link#add（Object，Object）}添加值，否则新的累加器通常没有意义。
      *
      * <p>The accumulator is the state of a running aggregation.
      * 累加器是正在运行的聚合的状态。
      * When a program has multiple aggregates in progress (such as per key and window), the state (per key and window) is the size of the accumulator.
      * 当一个程序有多个正在进行的聚合（例如每个键和窗口）时，状态（每个键和窗口）是累加器的大小。
      *
      * @return A new accumulator, corresponding to an empty aggregate. 返回一个新的累加器，对应一个空的聚合。
      */
    override def createAccumulator(): Long = 0L


    /**
      * Gets the result of the aggregation from the accumulator.
      * 从累加器获取聚合的结果。
      *
      * @param accumulator The accumulator of the aggregation  聚合的累加器
      * @return The final aggregation result.  返回最终聚合结果。
      */
    override def getResult(accumulator: Long): Long = {
      println("【getResult】accumulator: "+ accumulator)
      accumulator
    }

    /**
      * Merges two accumulators, returning an accumulator with the merged state.
      * *合并两个累加器，返回合并状态的累加器。
      *
      * <p>This function may reuse any of the given accumulators as the target for the merge
      * and return that.
      * 此函数可以重用任何给定的累加器作为合并的目标，并返回该累加器。
      *
      * The assumption is that the given accumulators will not be used any more after having been passed to this function.
      * 假设给定的蓄能器在传递给此函数后将不再使用。
      *
      *
      * @param a An accumulator to merge  要合并的累加器
      * @param b Another accumulator to merge   另一个要合并的累加器
      * @return The accumulator with the merged state   返回合并状态的累加器
      */
    override def merge(a: Long, b: Long): Long = {
      //println("【merge】(a + b): a"+ a+" ,b: "+b)
      a + b
    }

  }


  // 自定义预聚合函数计算平均数： 计算UserBehavior的timestamp字段的平均值
  class AverageAgg() extends AggregateFunction[/*输入数据类型*/UserBehavior, /*定义元祖作为状态类型*/ ( /*timestamp时间戳用Long*/Long, /*timestamp时间戳的数量*/ Int), /*输出数据类型，平均值用Double*/Double]{

    override def add(value: UserBehavior, accumulator: (Long, Int)): (Long, Int) = {
      /*返回元祖*/
      println("【add】 accumulator._1: " + accumulator._1+ " , value.timestamp： "+ value.timestamp +" , accumulator._2 + 1: " + accumulator._2 + 1)
      (accumulator._1 +  value.timestamp ,accumulator._2 + 1)
    }

    override def createAccumulator(): (Long, Int) = {
      // 返回初始值元祖
      (0L ,0)
    }

    override def getResult(accumulator: (Long, Int)): Double = {
      // 返回平均值
      println("【getResult】 accumulator._1: " + accumulator._1+" , getResult accumulator._2: " + accumulator._1)
      (accumulator._1 / accumulator._2)
    }

    override def merge(a: (Long, Int), b: (Long, Int)): (Long, Int) = {
      // 两个累加器怎么合并到一起，仅仅是状态的改变，没有要算最后的结果
      println("【merge】 a._1: " + a._1+" , merge b._1: " + b._1+" , merge a._2: " + a._2+", merge b._2: " + b._2)
      (a._1 + b._1 , a._2 + b._2)
    }

  }



  // 自定义排序处理函数
  /**
    * KeyedProcessFunction
    *
    *处理流元素的键控函数。
    *
    *<p>对于输入流中的每个元素，{@link#processElement（Object，Context，Collector）}被调用。
    *
    * 这可以产生零个或多个元素作为输出。实现还可以通过提供的{@linkcontext}查询时间并设置计时器。
    *
    * 对于触发计时器，将调用{@link#onTimer（long，OnTimerContext，Collector）}。
    *
    * 这可以再次产生零个或更多的元素作为输出，并注册更多的定时器。
    *
    * *<p><b>注意：</b>只有在{@code KeyedStream}上应用了{@code KeyedProcessFunction}时，才能访问键控状态和计时器（也限定于键）。
    *
    *<p><b>注意：</b>一个{@code-KeyedProcessFunction}总是一个{@link网址：apache.flink.api.common.functions函数.RichFunction}。
    *
    * 因此，访问{@link网址：apache.flink.api.common.functions函数.RuntimeContext}始终可用，可以实现setup和teardown方法。
    *
    * 参见{@link网址：apache.flink.api.common.functions函数.RichFunction#打开(网址：apache.flink.配置。配置)}和{@link网址：apache.flink.api.common.functions函数.RichFunction#close（）}。
    *
    *@param<K>键的类型。
    *@param<I>输入元素的类型。
    *@param<O>输出元素的类型。
    */
  class TopNHotItem( topSize:Int ) extends KeyedProcessFunction[ /*<K>键的类型  ,【由于 .keyBy(_.windowEnd)  中 windowEnd是Long类型，所以这里K也是Long类型】*/ Long, /*<I>输入元素的类型 【当前输入的状态就是 ItemViewCount】*/  ItemViewCount , /*<O>输出元素的类型 【输出直接在控制台打印String就行】*/ String ]{

    // 【初始化为空值：  = _ 】
    private var itemState: ListState[ ItemViewCount ] = _

    /**
      * 函数的初始化方法。
      *
      * 它在实际工作方法（如map或join）之前调用，因此适用于一次性设置工作。
      *
      * 对于作为迭代一部分的函数，此方法将在每个迭代步骤的开始处调用。
      * @param parameters
      */
    override def open(parameters: Configuration): Unit = {
      /* 默认实现： super.open(parameters)*/
      // 初始化 itemState
      itemState = getRuntimeContext.getListState( new ListStateDescriptor[ItemViewCount](/*name: String*/ "item-state" , /*typeSerializer: TypeSerializer[T]*/ classOf[ ItemViewCount ] ))
    }


    override def processElement(value: ItemViewCount, ctx: KeyedProcessFunction[Long, ItemViewCount, String]#Context, out: Collector[String]): Unit = {
      // 处理流程：来一条数据就存放在状态里面先保存起来，然后注册定时器，等到所有的数据都到齐了触发定时器进行排序输出
      // 实际上就是把每条数据都存起来

      // - step 1: 先定义状态
      // 在 open 方法中进行初始化

      // - step 2: 把每条状态都存起来
      itemState.add(value)

      // - step 3: 注册一个定时器  【这里需要的 EventTimer 定时器， 靠 waterMark 触发的 】
      // 【 value.windowEnd + 1 : 延迟 value.windowEnd + 1 毫秒】
      ctx.timerService().registerEventTimeTimer( value.windowEnd + 1 )

      // - step 4: 真正核心的操作要在定时器触发的时候操作，即 onTimer，所以要实现 onTimer接口
    }

    // 定时器触发，对所有数据排序， 并输出结果
    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, ItemViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {
      /* 默认实现：super.onTimer(timestamp, ctx, out)*/

      // - step 1:  将所有state中的数据取出，放到一个List Buffer中
      var allItems: ListBuffer[ ItemViewCount ] = new ListBuffer[ItemViewCount]()

      // - step 2: 要把itemState中所有的数据遍历处理，放到allItems中，
      // 注意，如果要做类似 foreach遍历的话，需要import一个转换，如下
      import scala.collection.JavaConversions._
      for ( item <- itemState.get() ){
        allItems += item
      }

      // - step 3: 按照点击量即count大小进行排序，并取前N个
      // 注意： sortBy 默认是升序  ，如果要降序，加上参数： Ordering.Long.reverse
      // 注意【.sortBy( _.count )( Ordering.Long.reverse ) 是函数的柯里化表达】 ，什么是柯里化 ？ https://www.jianshu.com/p/2975c25e4d71
      val sortedItems = allItems.sortBy( _.count )( Ordering.Long.reverse ) // 按照count降序排序
        // topN：用take即可
        .take(topSize) // topN

      // - step 4: 清空状态：把itemState清空，释放内存空间
      itemState.clear()

      // - step 5: 格式化输出 ：将排名结果输出
      val result: StringBuilder = new StringBuilder()
      result.append("当前窗口时间： ")
        .append( new Timestamp( /*为什么这里timestamp减去1，因为前面value.windowEnd加1，这里减1恢复 */ timestamp -1 ))
        .append("\n")
      // 输出每一个商品的信息
      // 【遍历写法1】【不推荐】 for ( i <- 0 to sortedItems.length -1 )
      // 【遍历写法2】【推荐】
      for ( i <- sortedItems.indices ){
        val  currentItem = sortedItems(i)
        result.append("No").append( i +1 ).append(": ")
          .append("商品Id= ").append(currentItem.itemId)
          .append("浏览量= ").append(currentItem.count)
          .append("\n")
      }
      result.append("======================")

      // 控制输出频率
      Thread.sleep(1000)

      // 输出
      out.collect(result.toString())
    }

  }



}
